package cn.wolfcode.rockitmq.producer.controller;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;

@RestController
public class DemoController {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    //发送消息
    @GetMapping("/send")
    public String send(String msg) {
        SendResult sendResult = rocketMQTemplate.syncSend("01-boot", msg);
        System.out.println(sendResult.getSendStatus());
        return "发送成功！";
    }

    //同步消息
    @GetMapping("/syncsend")
    public String syncsend(String msg) {
        SendResult sendResult = rocketMQTemplate.syncSend("02-boot", msg);
        System.out.println(sendResult.getSendStatus());
        return "发送同步消息成功！";
    }

    //异步消息
    @GetMapping("/asyncsend")
    public String asyncsend(String msg) {
        rocketMQTemplate.asyncSend("03-boot", msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println(sendResult.getSendStatus());
                System.out.println("消息消费成功");
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                System.out.println("消息消费失败");
            }
        });
        return "发送异步消息成功！";
    }


    //一次性消息
    @GetMapping("/onewaysend")
    public String onewaysend(String msg) {
        rocketMQTemplate.sendOneWay("04-boot", msg);
        return "发送一次性消息成功！";
    }


    //消费模式：集群和广播
    @GetMapping("/consumermode")
    public String consumermode(String msg) {
        SendResult sendResult = rocketMQTemplate.syncSend("05-boot", msg);
        System.out.println(sendResult.getSendStatus());
        return "发送测试消费模式的信息成功！";
    }

    //延时消息
    @RequestMapping("/delaysend")
    public String delaysend(String msg) throws Exception {
        //前提: 需要将参数中的 message 构建好
        //MessageBuilder<String> builder = MessageBuilder.withPayload(msg);
        //org.springframework.messaging.Message<String> message = builder.build();

        //方式1：设置延时消息的方式
        //SendResult sendResult = rocketMQTemplate.syncSend("06-boot", message, 5000, 3);

        //方式2：使用原生的API来实现
        DefaultMQProducer producer = rocketMQTemplate.getProducer(); //直接拿到生产者进行操作
        Message message1 = new Message("06-boot", "原生API实现".getBytes());
        message1.setDelayTimeLevel(3);  //延迟等级
        SendResult sendResult = producer.send(message1);

        System.out.println(sendResult.getSendStatus());
        return "发送延时消息测试的信息成功！";
    }


    //设置消息标签    在发送的消息Topic:Tag 中间使用冒号隔开，可以自动解析
    @GetMapping("/settag")
    public String settag(String msg, String tag) {
        // Topic:Tag
        SendResult sendResult = rocketMQTemplate.syncSend("01-boot:" + tag, msg);
        System.out.println(sendResult.getSendStatus());
        return "发送成功！";
    }

    //自定义属性设置   消费者可以通过SQL29来过滤
    @GetMapping("/filtersend")
    public String filtersend(String msg, String tag, String age) {
        //先用map封装属性
        HashMap<String, Object> map = new HashMap<>();
        //自定义属性
        map.put("age", age);
        map.put("high", "180");

        //设计 key : 用 MessageConst 常量类中的 PROPERTY_KEYS 来设计
        map.put(MessageConst.PROPERTY_KEYS, "myselfey");

        // Topic:Tag
        rocketMQTemplate.convertAndSend("01-boot:" + tag, msg, map);

        return "发送成功a！";
    }

}
